package com.ms.education.statistics.handler;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.JobType;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobRootConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.internal.schedule.JobRegistry;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.ms.education.statistics.entity.JobConfigParameters;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.sql.DataSource;

/**
 * <p>Project:我的代码-ElasticJobHandler
 * <p>Powered by ms on 2023-02-11 16:46:06
 *
 * @author ms
 * @version 1.0
 * @since 1.8
 */
@Component
public class
ElasticJobHandler {

    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @Autowired
    private ElasticJobListener elasticJobListener;


    /**
     * 配置任务详细信息
     *
     * @return LiteJobConfiguration
     */
    private LiteJobConfiguration createJobConfiguration(final JobConfigParameters parameters) {
        // 定义作业核心配置
        JobCoreConfiguration.Builder coreConfigBuilder = JobCoreConfiguration.newBuilder(
                parameters.getJobName(),
                parameters.getCron(),
                parameters.getShardingTotalCount());
        if (!StringUtils.isEmpty(parameters.getShardingItemParameters())) {
            coreConfigBuilder.shardingItemParameters(parameters.getShardingItemParameters());
        }
        if (!StringUtils.isEmpty(parameters.getJobParameter())) {
            coreConfigBuilder.jobParameter(parameters.getJobParameter());
        }
        coreConfigBuilder.failover(parameters.isFailover())
                .misfire(parameters.isMisfire())
                .description(parameters.getDescription());
        JobCoreConfiguration jobCoreConfiguration = coreConfigBuilder.build();
        JobTypeConfiguration jobTypeConfiguration = null;
        // 定义SIMPLE类型配置
        if (parameters.getJobType().equals(JobType.SIMPLE)) {
            jobTypeConfiguration = new SimpleJobConfiguration(jobCoreConfiguration,
                    parameters.getJobClass().getCanonicalName());
        } else if (parameters.getJobType().equals(JobType.DATAFLOW)) {
            jobTypeConfiguration = new DataflowJobConfiguration(jobCoreConfiguration,
                    parameters.getJobClass().getCanonicalName(), true);
        }

        // 定义Lite作业根配置
        return LiteJobConfiguration
                .newBuilder(jobTypeConfiguration)
                .overwrite(true)
                .jobShardingStrategyClass("com.dangdang.ddframe.job.lite.api.strategy.impl.OdevitySortByNameJobShardingStrategy")
                .build();
    }

    //添加定时任务
    public void addJob(final JobConfigParameters parameters) {
//        JobType type = null;
//        if (parameters.getElasticJob() instanceof SimpleJob) {
//            type = JobType.SIMPLE;
//        }
//        if (parameters.getElasticJob() instanceof DataflowJob) {
//            type = JobType.DATAFLOW;
//        }
//        parameters.setJobType(type);
        LiteJobConfiguration jobConfiguration = createJobConfiguration(parameters);

        new SpringJobScheduler(parameters.getElasticJob(), zookeeperRegistryCenter, jobConfiguration, elasticJobListener).init();
    }

    /**
     * 修改定时任务
     *
     * @param jobName 任务名称
     * @param cron    运行机制
     */
    public void updateJob(String jobName, String cron) {
        JobRegistry.getInstance().getJobScheduleController(jobName).rescheduleJob(cron);
    }

    /**
     * 移除定时任务
     *
     * @param jobName 任务名称
     */
    public void removeJob(String jobName) {
        JobRegistry.getInstance().getJobScheduleController(jobName).shutdown();
    }

}
